"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : security-overflow
Case Name   : 执行COMMENT SQL语句时没有对commen 参数长度进行限制，测试指定过长的comment是否会造成缓冲区溢出
Create At   : 2023/06/05
@zou_jialiang0501162244
Description :
    1. 创建测试表
    2. 给测试表中的列添加注释
    3. 显示测试表中的注释
    4. 清理环境
Expect      :
    1. 成功
    2. 成功
    3. 成功
    4. 成功
History     :
"""

import os
import unittest

from yat.test import Node
from testcase.utils.Logger import Logger
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant


class Overflow(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.pri_node = Node('PrimaryDbUser')
        self.cons = Constant()
        self.pri_sh = CommonSH('PrimaryDbUser')
        self.table = 't_memory_0009'

    def test_security(self):
        text1 = '-----step1:创建测试表;expect:成功-----'
        self.log.info(text1)
        msg = self.pri_sh.execut_db_sql(
            f'drop table if exists {self.table} cascade;'
            f'create table {self.table}(id int);')
        self.log.info(msg)
        self.assertIn(self.cons.CREATE_TABLE_SUCCESS, msg, f'执行失败:{text1}')

        text2 = '-----step2:给测试表中的列添加注释;expect:成功-----'
        self.log.info(text2)
        comment = 'a' * 16000
        msg = self.pri_sh.execut_db_sql(
            f"comment on column {self.table}.id is '{comment}';")
        self.log.info(msg)
        self.assertIn('COMMENT', msg, f'执行失败：{text2}')

        text3 = '-----step3:显示测试表中的注释;expect:成功-----'
        self.log.info(text3)
        msg = self.pri_sh.execut_db_sql(
            f"select description from pg_description d join pg_class c "
            f"on d.objoid = c.oid where relname = '{self.table}';")
        self.log.info(msg)
        self.assertIn(comment, msg, f'执行失败:{text3}')

    def tearDown(self):
        text4 = '-----step4:清理数据;expect:成功-----'
        self.log.info(text4)
        sql_cmd = f"drop table if exists {self.table} cascade;"
        msg = self.pri_sh.execut_db_sql(sql_cmd)
        self.log.info(msg)
        self.assertIn(self.cons.DROP_TABLE_SUCCESS, msg, f"执行失败:{text4}")
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
